#pragma once // 防止头文件被重复包含
#include <pthread.h>
#include <vector>
#include <sys/eventfd.h>
#include <unistd.h>  
#include <stdint.h>
#include <stdlib.h>  
#include <fcntl.h>
#include <cstdint> 
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <assert.h>
#include <errno.h> 
#include <unistd.h> 
#include <memory>
#include <assert.h>
#include <algorithm>
#include <string>
#include <cstring>
#include <iostream>
#include <string>
#include <ctime>
#include <sys/socket.h>
#include <cstdlib>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <string.h>
#include "log.hpp"
#include <sys/timerfd.h>
#include <any>

#define BACKLOG 1024 // 全连接队列最小值
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 
class tcp_serve
{
public:
  static tcp_serve *git_tcp_object(int port) // 获取单例实例
  {
    if (tcp_ptr == nullptr)
    {
      pthread_mutex_lock(&mutex); // 加锁
      if (tcp_ptr == nullptr)
      {
        tcp_ptr = new tcp_serve(port);
        tcp_ptr->initserve();
      }
      pthread_mutex_unlock(&mutex); // 解锁
    }
    return tcp_ptr;
  }
 
  int git_socket_fd()
  {
    return _socket_fd;
  }
private:
  void initserve()
  {
    Socket();
    Bind();
    Listen();
  }
  tcp_serve(int port)
      : _port(port), _socket_fd(-1)
  {
  }
  ~tcp_serve()
  {
  }
 
  tcp_serve(const tcp_serve &x);
  const tcp_serve &operator=(const tcp_serve &x);
  void Socket()
  {
    int fd = socket(AF_INET, SOCK_STREAM, 0); // 用IPv4协议家族，TCP协议
    if (-1 == fd)
    {
      LOG(FATAL, "创建套接字失败");
      exit(1);
    } // 创建套接字失败
 
    _socket_fd = fd;
  }
 
  void Bind()
  {
    struct sockaddr_in address;
    memset(&address, 0, sizeof(address)); // 把结构体初始化为0
 
    address.sin_family = AF_INET;         // IPv4协议家族
    address.sin_addr.s_addr = INADDR_ANY; // 服务器选择任意IP地址接收请求
    address.sin_port = htons(_port);      // 端口号信息,网络序列
 
    int opt = 1;
    if (setsockopt(_socket_fd, SOL_SOCKET, SO_REUSEADDR, (const void *)&opt, sizeof(opt)) < 0)
    {
      LOG(WARNING, "端口不能在短时间内连续绑定");
    } // 防止TIME_WAIT状态导致绑定端口失败
    if (-1 == bind(_socket_fd, (const struct sockaddr *)&address, (socklen_t)sizeof(address)))
    {
      LOG(FATAL, "绑定套接字失败");
      exit(2); // 绑定套接字失败
    }
  }
 
  void Listen()
  {
    if (-1 == listen(_socket_fd, BACKLOG))
    {
 
      LOG(FATAL, "声明自己可连接失败");
      exit(3);
    } // 请求链接失败
 
    LOG(INFO, "套接字允许监听链接");
  }
 
private:
  int _port;      // 端口
  int _socket_fd; // 套接字文件描述符
  static tcp_serve *tcp_ptr;
};
 
tcp_serve *tcp_serve::tcp_ptr = nullptr;


#define BUFF_MAX_SIZE 1024  //容器大小的初始值


class Buff{
    private:
    std::vector<char> _buff;    //缓冲区
    uint64_t _read_pos;         //读下标
    uint64_t _write_pos;        //写下标
    public:
    //获取容器起始地址
    char* GetInitPos(){ return &( *_buff.begin());   }
    //获取读位置地址
    char* GetReadPos(){ return GetInitPos() + _read_pos; }
    //获取写位置地址
    char* GetWritePos(){    return GetInitPos() + _write_pos;   }
    //缓冲区读位置之后有多少空闲空间,把空间分为了三段,第一段
    uint64_t FirstParagraphV(){     return _read_pos;   }
    //可读数据大小，第二段
    uint64_t SecondParagraphV(){    return _write_pos - _read_pos;  }
    //缓冲区写位置之前有多少空闲空间，第三段
    uint64_t ThirdParagraphV(){     return _buff.size() - _write_pos;   }
    //确保足够空间
    void EnsureSpaceEnough(uint64_t len){
        if (len <= ThirdParagraphV()){
            return;
        }       
        else if ( len <= FirstParagraphV() + ThirdParagraphV()){
            std::copy(GetReadPos(), GetReadPos() + SecondParagraphV(), GetInitPos());
            _write_pos = SecondParagraphV();
            _read_pos = 0;
        }
        else{   _buff.resize(_buff.size() + len);  }
    }
    public:
    //构造
    Buff():_read_pos(0), _write_pos(0), _buff(BUFF_MAX_SIZE){   }
    //可读数据大小
    uint64_t ReadSize(){    return SecondParagraphV();  }
    uint64_t size() { return _buff.size(); };
      //读偏移移动
    void MoveReadPos(uint64_t len){
        assert(len <= SecondParagraphV());
        _read_pos += len;
    }
    //写偏移移动
    void MoveWritePos(uint64_t len){
        assert(len <= ThirdParagraphV());
        _write_pos += len;
    }
    //写入数据
    void Write(const char* date, uint64_t len){
        EnsureSpaceEnough(len); //确保空间足够
        std::copy(date, date + len, GetWritePos());     //写入数据
        MoveWritePos(len);  //移动下标
    }
    void Write(const std::string& date, uint64_t len){  Write(&date[0], len);   }
    void Write(Buff& date, uint64_t len){   Write(date.GetReadPos(), len);  }
    void Write(int fd, uint64_t len) {
        std::string str;
        str.resize(len);
        read(fd, &str[0], len);
        Write(str, len);
    }
    //窥探数据
    void pry(char* buf, uint64_t len){
        assert(len <= SecondParagraphV());
        std::copy(GetReadPos(), GetReadPos() + len, buf);
    }
    void pry(std::string& buf, uint64_t len){
        assert(buf.size() >= SecondParagraphV());
        pry(&(*buf.begin()), len);
        }
    void pry(Buff& buf, uint64_t len){
        assert(buf.size() >= size());
        pry(buf.GetWritePos(), len);    
        }
    //读取数据
    void Read(char* buf, uint64_t len){
        assert(len <= SecondParagraphV());
        std::copy(GetReadPos(), GetReadPos() + len, buf);
        MoveReadPos(len);
    }
    void Read(std::string& buf, uint64_t len){
        assert(buf.size() >= SecondParagraphV());
        Read(&(*buf.begin()), len);
        }
    void Read(Buff& buf, uint64_t len){
        assert(buf.size() >= size());
        Read(buf.GetWritePos(), len);
        }
    //读取一行, \r \r\n \n
void ReadLen(std::string& str) {
   str.resize(SecondParagraphV());
   while (SecondParagraphV() > 0) {
     if (*GetReadPos() == '\r' || *GetReadPos() == '\n') {
        if (*GetReadPos() == '\r' && *(GetReadPos() + 1) == '\n') {
            str += *GetReadPos();
            str += *(GetReadPos() + 1);
            MoveReadPos(2);
            return;
        }
        str += *GetReadPos();
        MoveReadPos(1); 
        return;
     }
        str += *GetReadPos();
        MoveReadPos(1);
   }
}
//一行数据是否小于8kb
bool IsLenSize() {
    if (SecondParagraphV() <= 8192) { return true; }
    char* read_ptr = GetReadPos();
    for (int i = 0; i < 8192; i++) {
        if (read_ptr[i] == '\r' || read_ptr[i] == '\n') { return true; }
    }
    return false;
}


    //清空缓冲区
    void Clear(){
        _read_pos = 0;
        _write_pos = 0;
    }
};



class EventLoop;

class Channel {
    private:
        int _fd;
        EventLoop* _eventloop;
        uint32_t _events;  // 当前需要监控的事件
        uint32_t _revents; // 当前连接触发的事件
        using EventCallback = std::function<void()>;
        EventCallback _read_callback;   //可读事件被触发的回调函数
        EventCallback _write_callback;  //可写事件被触发的回调函数
        EventCallback _error_callback;  //错误事件被触发的回调函数
        EventCallback _close_callback;  //连接断开事件被触发的回调函数
        EventCallback _event_callback;  //任意事件被触发的回调函数
    public:
        Channel(int fd, EventLoop* p):_fd(fd), _events(0), _revents(0), _eventloop(p){}
        int GetFd() { return _fd; }
        void UpdateFd(int fd) { _fd = fd; }
        uint32_t Events() { return _events; }//获取想要监控的事件
        void SetREvents(uint32_t events) { _revents = events; }//设置实际就绪的事件
        void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
        void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
        void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
        void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
        void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
        //当前是否监控了可读
        bool ReadAble() { return (_events & EPOLLIN); } 
        //当前是否监控了可写
        bool WriteAble() { return (_events & EPOLLOUT); }
        //启动读事件监控
        void EnableRead() { _events |= EPOLLIN; Update(); }
        //启动写事件监控
        void EnableWrite() { _events |= EPOLLOUT; Update(); }
        //关闭读事件监控
        void DisableRead() { _events &= ~EPOLLIN; Remove(); }
        //关闭写事件监控
        void DisableWrite() { _events &= ~EPOLLOUT; Remove(); }
        //关闭所有事件监控
        void DisableAll() { _events = 0; Remove(); }
        void Update();
        void Remove();
        //事件处理，一旦连接触发了事件，就调用这个函数，自己触发了什么事件如何处理自己决定
        void HandleEvent() {
            //EPOLLIN 读事件  EPOLLRDHUP 对端关闭链接了  EPOLLPRI  带外数据（紧急数据）
            if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)) {
                if (_read_callback) _read_callback();
            }
            //有可能会释放连接的操作事件，一次只处理一个
            if (_revents & EPOLLOUT) {  //EPOLLOUT 写事件
                if (_write_callback) _write_callback();
            }else if (_revents & EPOLLERR) {    //EPOLLERR错误事件
                if (_error_callback) _error_callback();//一旦出错，就会释放连接，因此要放到前边调用任意回调
            }else if (_revents & EPOLLHUP) {
                if (_close_callback) _close_callback();
            }
            //不管任何事件，都调用的回调函数
            if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI) || (_revents & EPOLLOUT) || (_revents & EPOLLERR) ||(_revents & EPOLLHUP)) {
                if (_event_callback) _event_callback();
            }
        }
};

#define EPOLL_WAIT_MAX_SIZE 1024

class poller{
    private:
    int _epoll_fd;  //epoll模型的句柄（文件描述符）
    struct epoll_event _ev[EPOLL_WAIT_MAX_SIZE];     //epoll_wait要捞取的就绪事件
    std::unordered_map<int, Channel*> _map;     //文件描述符和channel的映射
    private:
    //向epoll模型中注册或向epoll模型中更改
    int EpollCtl(int op, int fd){ 
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = _map[fd]->Events();
        int ret = epoll_ctl(_epoll_fd, op, fd, &ev);
        if (-1 == ret) { LOG(WARNING, "向epoll模型的注册或更新操作失败"); }
        return ret;
    }
    //判断文件描述符是否在_map中, 该文件描述符是否被poller管理
    bool IfPollerManage(int fd){
        auto it = _map.find(fd);
        if (it == _map.end()){ return false; }
        return true;
    }
    public:
    poller(){
       _epoll_fd = epoll_create(EPOLL_WAIT_MAX_SIZE);
       if (-1 == _epoll_fd) {
        LOG(ERROR, "epoll模型创建失败");
        exit(1);
       }
    }
    ~poller(){ if (-1 == close(_epoll_fd)) { LOG(WARNING, "epoll模型销毁失败"); } }
    //更新or注册链接
    void Update(Channel* c_ptr){
        if (IfPollerManage(c_ptr->GetFd())){
            EpollCtl(EPOLL_CTL_MOD, c_ptr->GetFd());
            return;
        }
        _map[c_ptr->GetFd()] = c_ptr;
        EpollCtl(EPOLL_CTL_ADD, c_ptr->GetFd());
    }
        //移除链接
    bool Remove(Channel* c_ptr) {
        if (IfPollerManage(c_ptr->GetFd())) {
            if (-1 != EpollCtl(EPOLL_CTL_DEL, c_ptr->GetFd())) {
                 _map.erase(c_ptr->GetFd());
                 return true;
            }
           return false;
        }
        return false;
    }
    //从epoll模型中的就绪队列中捞取事件就绪的链接
    bool Wait(std::vector<Channel*> &v){
        int ret = epoll_wait(_epoll_fd, _ev, EPOLL_WAIT_MAX_SIZE, -1);
        if (-1 == ret){
            LOG(WARNING, strerror(errno));
            return false;
        }
        for (int i = 0; i < ret; i++) {
            auto it = _map.find(_ev[i].data.fd);
            assert (it != _map.end());
            _map[_ev[i].data.fd]->SetREvents((_ev[i]).events);
            v.push_back(_map[_ev[i].data.fd]);
        }
        return true;
    }
};




using TaskFunk = std::function<void()>;  // 定时任务执行的方法
using CleanFunk = std::function<void()>; // 清理资源的回调
class TimerTask
{
private:
    uint64_t _id;        // 任务对象的唯一性标识
    uint32_t _timeout_t; // 任务的超时时间
    bool _cancel;        // 取消任务为true, 不取消为false
    TaskFunk _taskfunk;  // 要执行的任务
    CleanFunk _cleanfunk;

public:
    TimerTask(uint64_t id, uint32_t timeout_t, const TaskFunk &funk)
        : _id(id), _timeout_t(timeout_t), _cancel(false), _taskfunk(funk)
    {
    }
    ~TimerTask()
    {
        if (_cancel == false)
        {
            _taskfunk();
        }
        _cleanfunk();
    }
    void AddCleanFunk(const CleanFunk &func)
    {
        _cleanfunk = func;
    }
    uint32_t GetTimeout() // 获取超时时间
    {
        return _timeout_t;
    }
    void CancelTask() // 取消任务
    {
        _cancel = true;
    }
};


class TimerWheel
{
private:
    using TaskSharedPtr = std::shared_ptr<TimerTask>;
    using TaskWeakPtr = std::weak_ptr<TimerTask>;
    size_t _ptr;                                    // 时间轮的指针
    size_t _capacity;                               // 时间轮的容量
    std::vector<std::vector<TaskSharedPtr>> _wheel; // 时间轮容器
    std::unordered_map<uint64_t, TaskWeakPtr> _v;   // 任务id和任务weak_ptr映射的容器，快速索引，使shared_ptr引用计数加一
    
    EventLoop* _loop;
    int _timerfd;           //定时器文件描述符
    Channel _channel;      //管理定时器文件描述符
    

private:
    // 清理_v容器资源的函数
    void CleanV(uint64_t id) 
    {
        auto t = _v.find(id);
        if (t != _v.end())
        {
            _v.erase(id);
        }
    }
    //获取定时器的文件描述符
    int CreateTimerFd() {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        struct itimerspec itm;
        itm.it_value.tv_sec = 1;//设置第⼀次超时的时间
        itm.it_value.tv_nsec = 0;
        itm.it_interval.tv_sec = 1;//第⼀次超时后，每隔多⻓时间超时
        itm.it_interval.tv_nsec = 0;
         timerfd_settime(timerfd, 0, &itm, NULL);//启动定时器
         return timerfd;
    }
    //定时器文件描述符的channel的读回调
    void ReadTimerFdChannel() {
        uint64_t tmp;
        int ret = read(_timerfd, &tmp, sizeof(tmp));
        if (ret < 0) {
        LOG(FATAL, "定时器文件描述符出问题了");
        exit(errno);
        }
        UpdatePtr();
    }
     // 每隔?时间执行一次，定时器的秒针
     void UpdatePtr()
    {
        _wheel[_ptr].clear();
        _ptr++;
        _ptr %= _capacity;
    }
     void AddTimerTask(uint64_t id, uint32_t timeout_t, const TaskFunk &funk) // 添加定时任务
    {   if (HasTaskID(id) == true) return; //如果定时任务存在，直接返回
        TaskSharedPtr sp(new TimerTask(id, timeout_t, funk));
        TaskWeakPtr wp = sp;
        _v[id] = wp;                                                // 向_v中注册
        sp->AddCleanFunk(std::bind(&TimerWheel::CleanV, this, id)); // 设置清理_v容器资源的回调
        size_t pos = (_ptr + sp->GetTimeout()) % _capacity;
        _wheel[pos].push_back(sp);
    }
    void UpdateTimerTask(uint64_t id) // 更新任务的超时的时间
    {
        auto t = _v.find(id);
        if (t != _v.end())
        {
            TaskSharedPtr sp = _v[id].lock();
            size_t pos = (_ptr + sp->GetTimeout()) % _capacity;
            _wheel[pos].push_back(sp);
        }
    }
    void CancelTask(uint64_t id) // 取消任务
    {
        auto t = _v.find(id);
        if (t != _v.end())
        {
            TaskSharedPtr sp = _v[id].lock();
            sp->CancelTask();
        }
    }
public:
    TimerWheel(EventLoop* loop)
        : _ptr(0), _capacity(60), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerFd()),_channel(_timerfd, _loop)
    {
        _channel.SetReadCallback(std::bind(&TimerWheel::ReadTimerFdChannel, this));
        _channel.EnableRead();
    }
     // 添加定时任务
     void AddT(uint64_t id, uint32_t timeout_t, const TaskFunk &funk);
    // 更新任务的超时的时间
    void UpdateT(uint64_t id);
    // 取消任务
    void CancelT(uint64_t id); 
    //判断任务是否存在，存在线程安全问题
    bool HasTaskID(uint64_t id) {
        auto t = _v.find(id);
        if (t != _v.end())
        {
           return true;
        }
        return false;
    }
};



class EventLoop{
    private:
    using _taskwork = std::function<void()>;
    pthread_t _id;                                                                  //EventLoop 绑定的线程id
    std::vector<_taskwork> _taskqueue;                                              //任务队列
    int _efd;                                                                       //用于事件通知（eventfd）
    poller _poller;                                                                 //epoll模型
    std::unique_ptr<Channel> _channelptr = std::make_unique<Channel>(_efd, this);   //管理eventfd描述符 
    pthread_mutex_t _mutex = PTHREAD_MUTEX_INITIALIZER;                             //保证任务队列线程安全的锁,要几把锁🔒？？？？
    TimerWheel _timer;                                                
    private:
    //运行任务池中的所有任务
    void RunTask(){
        std::vector<_taskwork> v;
        {
            pthread_mutex_lock(&_mutex);
            _taskqueue.swap(v);
            pthread_mutex_unlock(&_mutex);
        }
        
        for (auto& f: v){ f(); }
    }
    //eventfd的回调
    void EventRead(){
        uint64_t t = 0;
        read(_efd, &t, 8);
        RunTask();
    }
    void EventWait(){
        uint64_t t = 1;
        write(_efd, &t, 8);
    }
        //是否是EventLoop绑定的线程ID
    bool EventLoopThreadID() {
        int ret = pthread_equal(_id, pthread_self());
        return ret;
    }

    public:
    EventLoop():_timer(this){
        _id = pthread_self();   //让eventloop绑定一个线程ID
        _efd = eventfd(0, O_NONBLOCK|O_CLOEXEC);
        if (-1 == _efd) {
            LOG(FATAL,"eventfd文件描述符创建失败,不能有效的进行事件通知");
            exit(errno);
        }  
       _channelptr->SetReadCallback(std::bind(&EventLoop::EventRead, this));
       _channelptr->EnableRead();
    }
    void AssertInLoop() {
        assert(EventLoopThreadID());
    }
    //把任务压入任务池
    void PushTask(const _taskwork& f) {
        {
            pthread_mutex_lock(&_mutex);
            _taskqueue.push_back(f);
            pthread_mutex_unlock(&_mutex);
        }
    }

    //如果是EventLoop的ID那么就执行任务，不是就压入任务池
    void RunEventLoopTask(const _taskwork & f) {
        if (EventLoopThreadID()) {
            return f();
        }
        //LOG(INFO, "不是当前线程");
        PushTask(f);
        EventWait();
    }
     //更新事件监控
    void Update(Channel* c);
    //移除事件监控
    void Remove(Channel* c);
    void Start() {
        std::vector<Channel*> v;
        _poller.Wait(v);
        for (auto& f : v) {
            f->HandleEvent();
        }
        RunTask();
    }
     // 添加定时任务
     void AddT(uint64_t id, uint32_t timeout_t, const TaskFunk &funk)
    {
        _timer.AddT(id, timeout_t, funk);
    }
    // 更新任务的超时的时间
    void UpdateT(uint64_t id) 
    {
        _timer.UpdateT(id);
    }
    // 取消任务
    void CancelT(uint64_t id) 
    {
        _timer.CancelT(id);
    }
    
};

typedef enum {  
    DISCONNECTED,  // 连接关闭状态  
    CONNECTING,    // 连接建立成功-待处理状态  
    CONNECTED,     // 连接建立完成，可以通信的状态  
    DISCONNECTING  // 待关闭状态  
} ConnStatu;     
// 使用智能指针定义连接类型  
using PtrConnection = std::shared_ptr<class Connection>; 
class Connection: public std::enable_shared_from_this<Connection>{
  public:
    uint64_t _conn_id;          //Connection唯一标识的ID
    int _socker_fd;             //通信套接字文件描述符
    bool _closedInactiveLink;   //是否关闭非活跃链接, true为关闭，false为不关闭
    ConnStatu _conn_statu;      //链接的状态
    Buff _in_buff;              //输入缓冲区
    Buff _out_buff;             //输出缓冲区
    EventLoop* _loop; 
    Channel _channel;           //对事件进行管理
    std::any _context;          //上下文


    // 回调函数类型定义  
    using ConnectedCallback = std::function<void(const PtrConnection&)>;  
    using MessageCallback = std::function<void(const PtrConnection&, Buff*)>;  
    using ClosedCallback = std::function<void(const PtrConnection&)>;  
    using AnyEventCallback = std::function<void(const PtrConnection&)>;  

    // 回调函数成员变量  
    ConnectedCallback _connected_callback;  
    MessageCallback _message_callback;      //业务处理的回调
    ClosedCallback _closed_callback;        //链接关闭回调
    AnyEventCallback _event_callback;       //触发任意事件的回调
    ClosedCallback _server_closed_callback; //组件内的链接关闭回调

    private:
    //链接的读实际回调
    void HandleRead() {
        char buf[65536];
        int ret = recv(_socker_fd, buf, 65535, MSG_DONTWAIT);   //MSG_DONTWAIT表示非阻塞
        if (-1 == ret) { ShutdownInLoop(); }                    //不能直接关闭链接链接
        _in_buff.Write(buf, ret);
        if (_in_buff.ReadSize() > 0) {
             _message_callback(shared_from_this(),&_in_buff);
            }  //业务处理回调
    }
    //链接的写事件监控
    void HandleWrite() {
        if (_out_buff.ReadSize() == 0) {
            if (_conn_statu == DISCONNECTING) {
                if (_in_buff.ReadSize() > 0) {
                _message_callback(shared_from_this(),&_in_buff); 
                }
                //关闭链接
                ReleaseInLoop();
                return;
            }
            _channel.DisableWrite();    //关闭写事件监控
            return;
        }
        std::string str;
        str.resize(_out_buff.ReadSize());
        _out_buff.pry(str, _out_buff.ReadSize());   //窥探数据
        int ret = send(_socker_fd, &str[0], _out_buff.ReadSize(), MSG_DONTWAIT);
        if (-1 == ret) {
            //可能对方把链接关了
                if (_in_buff.ReadSize() > 0) {
                _message_callback(shared_from_this(),&_in_buff); 
                ReleaseInLoop();
                }
            return;
        }
        _out_buff.MoveReadPos(ret); //移动发送缓冲区的读下标
    }
    //错误事件回调
    void HandleError() {
        HandleClose();
    }
    //关闭事件回调
    void HandleClose() {
        if (_in_buff.ReadSize() > 0) {
        _message_callback(shared_from_this(),&_in_buff); 
        ReleaseInLoop();
        }
    }
    //任意事件回调
    void HandleEvent() {
        if(_closedInactiveLink == false) {_loop->UpdateT(_conn_id); }    //刷新链接的活跃度
        if (_event_callback) { _event_callback(shared_from_this()); }
    }
    //启动链接,让链接变为可通信状态
    void EstablishedInLoop() {
        assert(_conn_statu == CONNECTING);
        _conn_statu = CONNECTED;
        _channel.EnableRead();              //启动读回调
        if (_connected_callback) (_connected_callback(shared_from_this()));
    }
    //关闭链接
    void ReleaseInLoop() {
        _conn_statu = DISCONNECTED;     //设置状态
        _channel.Remove();              //移除监控事件
        close(_socker_fd);              //关闭链接
        _loop->CancelT(_conn_id);       //移除定时任务
        if (_closed_callback) { _closed_callback(shared_from_this()); }
        if (_server_closed_callback) { _server_closed_callback(shared_from_this()); }
    }
    //关闭链接,并不直接关闭，要判断接收和发送缓冲区的数据释放是否需要处理
    void ShutdownInLoop() {
        _conn_statu = DISCONNECTING; //状态设置成待关闭状态
        if (_in_buff.ReadSize() > 0 ) {
            if (_message_callback) { _message_callback(shared_from_this(), &_in_buff); }
        }
        if (_out_buff.ReadSize() > 0) {
        if (_channel.WriteAble() == false) { _channel.EnableWrite(); } 
        }
        if (_out_buff.ReadSize() == 0) {
            _conn_statu = DISCONNECTED;
            Release();
        }
    }
    //把数据数据放到发送缓冲区，启动写事件监控
    void SendInLoop(Buff& buff) {
        if (_conn_statu == DISCONNECTED) return;
        _out_buff.Write(buff, buff.ReadSize());
        if (_channel.WriteAble() == false) { _channel.EnableWrite(); } 
    }
    //启动非活跃链接的销毁任务,//sec参数表示超时时间
    void EnableInactiveReleaseInLoop(uint64_t sec) {
        _closedInactiveLink = true;
        _loop->AddT(_conn_id, sec,std::bind(&Connection::ShutdownInLoop, this));
        _loop->UpdateT(_conn_id);
    }
    //关闭非活跃链接的销毁任务
    void CancelInactiveReleaseInLoop() {
        _closedInactiveLink = false;
        _loop->CancelT(_conn_id);
    }
    //更新协议，对数据的处理方式
            void UpgradeInLoop(const std::any &context, 
                    const ConnectedCallback &conn, 
                    const MessageCallback &msg, 
                    const ClosedCallback &closed, 
                    const AnyEventCallback &event) {
            _context = context;
            _connected_callback = conn;
            _message_callback = msg;
            _closed_callback = closed;
            _event_callback = event;
        }
    public:
    Connection(uint64_t id, int fd, EventLoop* loop):_conn_id(id), _closedInactiveLink(false),
    _conn_statu(CONNECTING),_socker_fd(fd), _loop(loop), _channel(_socker_fd, _loop) {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
        std::cout<<"创建了一个链接，链接的地址是："<<this<<std::endl;
    }
     ~Connection() {
            std::cout<<"链接被销毁了，链接的地址是："<<this<<std::endl;
    }

    //获取文件描述符
    int GetFd() { return _socker_fd; }
    //获取链接ID
    int GetID() { return _conn_id; }
    //判断该链接是否是可通信的状态
    bool Connected() { return (_conn_statu == CONNECTED); }
    //设置上下文
    void SetContext(const std::any &context) { _context = context; }
    //获取上下文，返回的是指针
    std::any *GetContextPtr() { return &_context; }
    std::any &GetContext() { return _context; }
    //设置回调
    void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback&cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback&cb) { _server_closed_callback = cb; }
    //让链接变为可通信的状态
    void Established() {
        _loop->RunEventLoopTask(std::bind(&Connection::EstablishedInLoop, this));
    }
    //发送数据，将数据放到发送缓冲区，启动写事件监控
    void Send(const char *data, size_t len) {
        Buff buf;
        buf.Write(data, len);
        _loop->RunEventLoopTask(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }
    //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
    void Shutdown() {
        _loop->RunEventLoopTask(std::bind(&Connection::ShutdownInLoop, this));
    }
    void Release() {
        _loop->PushTask(std::bind(&Connection::ReleaseInLoop, this));
    }
    //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    void EnableInactiveRelease(int sec) {
        _loop->RunEventLoopTask(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    //取消非活跃销毁
    void CancelInactiveRelease() {
        _loop->RunEventLoopTask(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    //切换协议---重置上下文以及阶段性回调处理函数 -- 而是这个接口必须在EventLoop线程中立即执行
    //防备新的事件触发后，处理的时候，切换任务还没有被执行--会导致数据使用原协议处理了。
    void Upgrade(const std::any &context, const ConnectedCallback &conn, const MessageCallback &msg, 
                 const ClosedCallback &closed, const AnyEventCallback &event) {
        _loop->AssertInLoop();
        _loop->RunEventLoopTask(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};

class Acceptor {
    private:
    tcp_serve* _tcp_serve;
    EventLoop* _loop;
    Channel _chann;
    //处理通信套接字的回调
    using ConnelFock = std::function<void(int)>;
    ConnelFock _connelfock;
    private:
    //监听套接字的读回调
    void ListeningRead(int lisfd) {
        struct sockaddr_in socket_c;
        socklen_t tmp = sizeof(socket_c);       // 结构体大小
        memset(&socket_c, 0, sizeof(socket_c)); // 把结构体初始化为0
        int fd = accept(lisfd, (struct sockaddr *)&socket_c, &tmp);
        if (_connelfock) { _connelfock(fd); }
    }
    public:
    Acceptor(int poot, EventLoop* loop):_tcp_serve(tcp_serve::git_tcp_object(poot)), _loop(loop), _chann(_tcp_serve->git_socket_fd(), _loop) {
        _chann.SetReadCallback(std::bind(&Acceptor::ListeningRead, this, _tcp_serve->git_socket_fd()));
        _chann.EnableRead();
    }
    void SetReadCallback(const ConnelFock & c) { _connelfock = c; }
    
};

//线程
class LoopTread{
    // 1.EventLoop的实例化要放到线程的执行流中 2.获取_loop指针要用互斥量和条件变量保护起来，避免外面拿到的是空指针
    private:
    pthread_t _id;              //线程ID
    pthread_mutex_t * _mutex;   //互斥量
    pthread_cond_t * _cond;     //条件变量
    EventLoop* _loop;           //线程关联的EventLoop
    //线程的执行流
    static void* ThreadFlow(void* arg);
    public:
    LoopTread():_id(-1), _mutex(new pthread_mutex_t), _cond(new pthread_cond_t), _loop(nullptr) {
        if (0 != pthread_mutex_init(_mutex, nullptr)) { LOG(WARNING, "互斥量初始化出错，可能会访问到空指针"); }
        if (0 != pthread_cond_init(_cond, nullptr)) { LOG(WARNING, "条件变量初始化出错，可能会访问到空指针"); }
        if (0 != pthread_create(&_id, nullptr, ThreadFlow, (void*)this)) {
            pthread_cancel(_id);
            LOG(ERROR, "线程创建失败，无法进行事件监控");
            return;
        }
        //线程分离
        if (0 != pthread_detach(_id)) { LOG(WARNING, "线程分离失败，可能会造成内存泄漏"); }
    }
    ~LoopTread() {
        pthread_mutex_destroy(_mutex);
        pthread_cond_destroy(_cond);
    }
    //获取_loop指针
    EventLoop* GetLoop() {
        EventLoop* loop;
        pthread_mutex_lock(_mutex);
        while (_loop == nullptr) {
            pthread_cond_wait(_cond, _mutex);
        }
        loop = _loop;
        pthread_mutex_unlock(_mutex);
        return loop;
    }

};

//线程池
class LoopTreadPool{
    private:
    uint32_t _size_max;                 //线程最大数量
    uint32_t _pos;                      //下标索引
    EventLoop* _baseloop;               //主loop
    std::vector<LoopTread*> _treads;    //从loop的线程池
    std::vector<EventLoop*> _loops;     //loop池
    public:
    LoopTreadPool(EventLoop* baseloop):_size_max(0), _pos(0), _baseloop(baseloop) {}
    //创建线程
    void CreadTread(uint32_t size = 0) {
        if (size > 0) {
            _size_max = size;
            _treads.resize(_size_max);
            _loops.resize(_size_max);
            for (int i = 0; i < _size_max; i++) {
                _treads[i] = (new LoopTread());
                _loops[i] = (_treads[i]->GetLoop());

            }
        }
    }
    EventLoop *NextLoop() {
        if (0 == _size_max) { return _baseloop; }
        _pos = (_pos + 1) % _size_max;
        return _loops[_pos];
    }
};

class TcpServe{
    private:
    int _poot;                      //端口
    uint64_t _conn_id;                   //链接ID
    bool _eir;                      //是否要启动非活跃链接销毁，默认false
    int _tread_size;                //从属线程的数量
    uint64_t _timeout;              //非活跃链接多长时间销毁
    EventLoop _baseloop;            //主loop,用于事件监听
    Acceptor _acceptor;             //监听链接
    LoopTreadPool _treadpool;       //从属线程池
    std::unordered_map<uint64_t, PtrConnection> _conn_map;  //管理链接
    // 回调函数类型定义  
    using ConnectedCallback = std::function<void(const PtrConnection&)>;  
    using MessageCallback = std::function<void(const PtrConnection&, Buff*)>;  
    using ClosedCallback = std::function<void(const PtrConnection&)>;  
    using AnyEventCallback = std::function<void(const PtrConnection&)>;  
    using ConnelFock = std::function<void()>;

    // 回调函数成员变量  
    ConnectedCallback _connected_callback;  
    MessageCallback _message_callback;      //业务处理的回调
    ClosedCallback _closed_callback;        //链接关闭回调
    AnyEventCallback _event_callback;       //触发任意事件的回调
    ClosedCallback _server_closed_callback; //组件内的链接关闭回调
   
    ConnelFock _fock;
    //添加定时任务
    void RunAfterInLoop(const ConnelFock& f, int sec) {
        _conn_id++;
        _baseloop.AddT(_conn_id, sec, f);
    }
    //为新链接创建Connection进行管理
    void NewConnection(int fd) {
        _conn_id++;
        Connection* conn = new Connection(_conn_id, fd, _treadpool.NextLoop());
        conn->SetAnyEventCallback(_event_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetMessageCallback(_message_callback);
        conn->SetSrvClosedCallback(_server_closed_callback);
        conn->Established();
        if (_eir == true) { conn->EnableInactiveRelease(_timeout); }
        PtrConnection p(conn);
        _conn_map[_conn_id] = p; 
    }
    //移除链接
    void RemoveConnectionInLoop(const PtrConnection &conn) {
        uint64_t id = conn->GetID();
        auto t = _conn_map.find(id);
        if (t != _conn_map.end()) {
            _conn_map.erase(id);
        }
    }
    //移除链接,线程安全
    void RemoveConnection(const PtrConnection& conn) {
        _baseloop.RunEventLoopTask(std::bind(&TcpServe::RemoveConnectionInLoop, this, conn));
    }
    public:
    TcpServe(int poot):_poot(poot), _conn_id(0), _eir(false),_tread_size(0),_timeout(0), _acceptor(_poot, &_baseloop), _treadpool(&_baseloop){
        _acceptor.SetReadCallback(std::bind(&TcpServe::NewConnection, this, std::placeholders::_1));
    }
    //线程数量
    void SetThreadSize(int size){
        _tread_size = size;
    }
    //启动非活跃链接定时销毁任务
    void SetTimeout(int sec) {
        _timeout = sec;
        _eir = true;
    }
    //设置回调
    void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback&cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback&cb) { _server_closed_callback = cb; }
    //添加定时任务
    void RunAfter(const ConnelFock& f, int sec) {
        _baseloop.RunEventLoopTask(std::bind(&TcpServe::RunAfterInLoop, this, f, sec));
    }
    void Start() {
         _treadpool.CreadTread(_tread_size);
        while (true) {
            _baseloop.Start();
        }
    }
};



pthread_mutex_t LoopTreadMutex = PTHREAD_MUTEX_INITIALIZER;
void* LoopTread::ThreadFlow(void* arg) {
    pthread_mutex_lock(&LoopTreadMutex);
    LoopTread* t_ptr =  (LoopTread*)arg;
    EventLoop loop;
    t_ptr->_loop = &loop;
   // std::cout<<"1000:"<<"我是线程："<<pthread_self()<<"我的loop是："<<&loop<<std::endl;
    pthread_cond_signal(t_ptr->_cond);
    pthread_mutex_unlock(&LoopTreadMutex);
    
    while (true) {
        loop.Start();
    }
}

void Channel::Remove() { return _eventloop->Remove(this); }
void Channel::Update() { return _eventloop->Update(this); }

 //更新事件监控
    void EventLoop::Update(Channel* c) {
        _poller.Update(c);
    }
    //移除事件监控
    void EventLoop::Remove(Channel* c) {
        _poller.Remove(c);
    }
 // 添加定时任务
     void TimerWheel::AddT(uint64_t id, uint32_t timeout_t, const TaskFunk &funk)
    {
        _loop->RunEventLoopTask(std::bind(&TimerWheel::AddTimerTask,this, id, timeout_t, funk));
    }
    // 更新任务的超时的时间
    void TimerWheel::UpdateT(uint64_t id) 
    {
        _loop->RunEventLoopTask(std::bind(&TimerWheel::UpdateTimerTask,this, id));
    }
    // 取消任务
    void TimerWheel::CancelT(uint64_t id) 
    {
        _loop->RunEventLoopTask(std::bind(&TimerWheel::CancelTask, this, id));
    }